package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.common.cluster.BrokerId;
import com.aiwiown.snackmq.common.dto.console.BrokerInfoDTO;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.serialization.JsonSerializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class ListBrokersHandler extends AbstractLeaderRequiredConsoleHandler {

    private final SnackMQRaftNodeManager dataRaftNodeManager;
    private final JsonSerializer serializer = SerializerFactory.getJsonSerializer();

    public ListBrokersHandler(MetadataManager metadataManager, SnackMQRaftNodeManager dataRaftNodeManager, AuthorizationService authorizationService) {
        super(metadataManager, authorizationService);
        this.dataRaftNodeManager = dataRaftNodeManager;
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.LIST_BROKERS_REQUEST;
    }

    @Override
    protected void doHandle(RequestContext ctx, Message request) {
        try {
            List<BrokerId> allBrokers = metadataManager.getAllBrokers();
            String metaLeaderAddress = metadataManager.getLeaderAddress();
            String dataLeaderAddress = dataRaftNodeManager.getLeaderAddress();

            // 添加调试信息
            log.info("ListBrokersHandler - All brokers: {}", allBrokers);
            log.info("ListBrokersHandler - Meta leader address: {}", metaLeaderAddress);
            log.info("ListBrokersHandler - Data leader address: {}", dataLeaderAddress);
            allBrokers.forEach(brokerId -> {
                log.info("BrokerId toString: '{}', getHost: '{}', getPort: {}", 
                         brokerId.toString(), brokerId.getHost(), brokerId.getPort());
            });

            List<BrokerInfoDTO> brokerInfos = allBrokers.stream().map(brokerId -> {
                BrokerInfoDTO dto = new BrokerInfoDTO();
                String brokerAddress = brokerId.toString();
                dto.setAddress(brokerAddress);
                String normalizedBrokerAddress = brokerId.getHost() + ":" + brokerId.getPort();
                dto.setMetaLeader(normalizedBrokerAddress.equals(metaLeaderAddress));
                dto.setDataLeader(normalizedBrokerAddress.equals(dataLeaderAddress));
                log.debug("Broker {} (normalized: {}) - metaLeader: {}, dataLeader: {}", 
                         brokerAddress, normalizedBrokerAddress, dto.isMetaLeader(), dto.isDataLeader());
                log.debug("Comparing with - metaLeaderAddress: {}, dataLeaderAddress: {}", 
                         metaLeaderAddress, dataLeaderAddress);
                return dto;
            }).collect(java.util.stream.Collectors.toList());

            Message response = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.SUCCESS)
                    .body(serializer.serialize(brokerInfos))
                    .build();
            ctx.writeResponse(response);

        } catch (Exception e) {
            log.warn("Failed to list brokers: {}", e.getMessage());
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage(e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        }
    }
}